# Copyright (C) 2021 - 2025 ANSYS, Inc. and/or its affiliates.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from collections import OrderedDict
import os
from pathlib import Path
import shutil
import tempfile

import pytest
from test_utils import pytest_approx

import ansys.fluent.core as pyfluent
from ansys.fluent.core import examples
from ansys.fluent.core.docker.utils import get_grpc_launcher_args_for_gh_runs
from ansys.fluent.core.filereader.case_file import CaseFile
from ansys.fluent.core.utils.fluent_version import FluentVersion


@pytest.mark.nightly
@pytest.mark.codegen_required
@pytest.mark.fluent_version("latest")
def test_simple_solve(mixing_elbow_param_case_data_session):
    """Use case 1: This optiSLang integration test performs these steps.

    - Reads a case file with and without data file
    - Gets input and output parameters and creates dictionary
    - Sets a variation on input parameters
    - Solve
    - Reread data

    This test queries the following using PyTest:
    - Session health
    - Input parameters
    - Output parameters
    """
    # Step 1: Launch fluent session and read case file with and without data file
    solver_session = mixing_elbow_param_case_data_session
    assert solver_session.is_server_healthy()
    if not solver_session.connection_properties.inside_container:
        solver_session.chdir(pyfluent.config.examples_path)
    case_name = "elbow_param.cas.h5"
    solver_session.settings.file.read(file_type="case", file_name=case_name)

    # Step 2: Get input and output parameters and create a dictionary
    reader = CaseFile(case_file_name=examples.path(case_name))

    input_parameters = {}
    for p in reader.input_parameters():
        input_parameters[p.name] = (p.value, p.numeric_value, p.units)
    output_parameters = {}
    for o in reader.output_parameters():
        output_parameters[o.name] = (0, o.units)
    solver_session.settings.file.read_case(file_name=case_name)

    input_parameters = input_parameters["inlet2_temp"]
    output_parameters = output_parameters["outlet_temp-op"]

    assert output_parameters[1] == "K"

    # Step 3: Set a variation on these input parameters
    # variations/designs are generated by optiSLang based on
    # algorithm selected
    fluent_input_table = solver_session.settings.parameters.input_parameters.expression[
        "inlet2_temp"
    ] = {"value": 600}

    tmp_save_dir = Path(tempfile.mkdtemp(dir=pyfluent.config.examples_path)).parts[-1]
    design_elbow_param_path = Path(tmp_save_dir) / "design_elbow_param.cas.h5"

    solver_session.settings.file.write_case(file_name=str(design_elbow_param_path))

    assert (Path(pyfluent.config.examples_path) / design_elbow_param_path).exists()

    # Step 4: Solve
    solver_session.settings.solution.initialization.standard_initialize()

    # check if solution is steady or transient
    workflow = solver_session.rp_vars("rp-unsteady?")

    # iterate workflow
    if workflow:
        solver_session.settings.solution.run_calculation.dual_time_iterate()
    else:
        solver_session.settings.solution.run_calculation.iterate()

    convergence = solver_session.rp_vars("solution/converged?")

    # solution output (test conditional statement)
    if not convergence:  # -> let user know
        print("Failed to converge")
    else:
        print("Solution is converged")

    assert convergence, "Solution failed to converge"

    # Step 5: Read the data again from the case and data file
    solver_session.settings.file.read_case_data(file_name=case_name)

    fluent_version = solver_session.get_fluent_version()
    fluent_input_table = solver_session.settings.parameters.input_parameters.list()
    for key, entry in fluent_input_table.items():
        input_value = entry[0]
        if fluent_version >= FluentVersion.v261:
            input_unit = list(entry[1].values())[0][-1]
        else:
            input_unit = entry[1]
    assert input_value == 500
    assert input_unit == "K"

    fluent_output_table = solver_session.settings.parameters.output_parameters.list()
    for key, entry in fluent_output_table.items():
        output_value = entry[0]
        if fluent_version >= FluentVersion.v261:
            output_unit = list(entry[1].values())[0][-1]
        else:
            output_unit = entry[1]

    assert output_value == pytest_approx(322.3360076327905)

    # output_unit should assert the unit string but it doesn't currently
    # A bug has been submitted to address this
    assert output_unit == "K"


@pytest.mark.nightly
@pytest.mark.codegen_required
@pytest.mark.fluent_version("latest")
def test_simple_solve_unitless(static_mixer_params_unitless_session):
    solver_session = static_mixer_params_unitless_session
    fluent_version = solver_session.get_fluent_version()

    case_name = "Static_Mixer_Parameters_unitless.cas.h5"
    reader = CaseFile(case_file_name=examples.path(case_name))

    input_parameters = {}
    for p in reader.input_parameters():
        input_parameters[p.name] = (p.value, p.numeric_value, p.units)
    output_parameters = {}
    for o in reader.output_parameters():
        output_parameters[o.name] = (0, o.units)

    assert input_parameters["inlets_v_ratio"] == ("2", 2.0, "")
    assert output_parameters["outlet-temp-uniformity-op"] == (0, "")

    solver_session.settings.parameters.input_parameters.expression["inlets_v_ratio"] = {
        "value": 3.0
    }

    fluent_input_table = solver_session.settings.parameters.input_parameters.list()
    input_parameters_dict = OrderedDict()
    for key, entry in fluent_input_table.items():
        input_parameters_dict[key] = OrderedDict()
        input_parameters_dict[key]["numeric_value"] = entry[0]
        if fluent_version >= FluentVersion.v261:
            input_parameters_dict[key]["units"] = list(entry[1].values())[0][-1]
        else:
            input_parameters_dict[key]["units"] = entry[1]

    assert input_parameters_dict["inlets_v_ratio"]["numeric_value"] == 3
    assert input_parameters_dict["inlets_v_ratio"]["units"] == "%"

    fluent_output_table = solver_session.settings.parameters.output_parameters.list()

    initialization_type = (
        solver_session.settings.solution.initialization.initialization_type()
    )
    if initialization_type == "hybrid":
        solver_session.settings.solution.initialization.hybrid_initialize()
    elif initialization_type == "standard":
        solver_session.settings.solution.initialization.standard_initialize()
    solver_session.settings.solution.run_calculation.iterate()

    output_parameters_dict = OrderedDict()
    fluent_output_table = solver_session.settings.parameters.output_parameters.list()
    for key, entry in fluent_output_table.items():
        output_parameters_dict[key] = OrderedDict()
        output_parameters_dict[key]["numeric_value"] = entry[0]
        output_parameters_dict[key]["units"] = entry[1]

    assert output_parameters_dict["outlet-temp-uniformity-op"][
        "numeric_value"
    ] == pytest_approx(0.9990138633523943)
    if fluent_version >= FluentVersion.v261:
        assert output_parameters_dict["outlet-temp-uniformity-op"]["units"] == [False]
    else:
        assert output_parameters_dict["outlet-temp-uniformity-op"]["units"] == ""


@pytest.mark.nightly
@pytest.mark.codegen_required
@pytest.mark.fluent_version("latest")
def test_generate_read_mesh(mixing_elbow_geometry_filename):
    """Use case 2: This optiSLang integration test performs these steps.

    - Launch Fluent in Meshing Mode
    - Generate mesh with default workflow settings
    - Read created mesh file
    - Switch to solution and write case file

    This test queries the following using PyTest:
    - Session health
    """
    # Step 1: Launch fluent session in meshing mode
    grpc_kwds = get_grpc_launcher_args_for_gh_runs()
    meshing_session = pyfluent.launch_fluent(
        mode="meshing", precision="double", processor_count=2, **grpc_kwds
    )
    assert meshing_session.is_server_healthy()
    if not meshing_session.connection_properties.inside_container:
        meshing_session.chdir(pyfluent.config.examples_path)
    temporary_resource_path = (
        Path(pyfluent.config.examples_path) / "test_generate_read_mesh_resources"
    )

    if os.path.exists(temporary_resource_path):
        shutil.rmtree(temporary_resource_path, ignore_errors=True)
    if not os.path.exists(temporary_resource_path):
        os.mkdir(temporary_resource_path)

    # Step 2: Generate mesh from geometry with default workflow settings
    meshing_session.workflow.InitializeWorkflow(WorkflowType="Watertight Geometry")
    geo_import = meshing_session.workflow.TaskObject["Import Geometry"]
    geo_import.Arguments = dict(FileName=mixing_elbow_geometry_filename)
    geo_import.Execute()
    meshing_session.workflow.TaskObject["Generate the Volume Mesh"].Execute()
    meshing_session.meshing.CheckMesh()
    temporary_resource_dir = Path(temporary_resource_path.parts[-1])
    gz_path = str(temporary_resource_dir / "default_mesh.msh.gz")
    h5_path = str(temporary_resource_dir / "default_mesh.msh.h5")
    meshing_session.meshing.File.WriteMesh(FileName=gz_path)
    meshing_session.meshing.File.WriteMesh(FileName=h5_path)
    assert (temporary_resource_path / "default_mesh.msh.gz").exists()
    assert (temporary_resource_path / "default_mesh.msh.h5").exists()

    # Step 3: use created mesh file - .msh.gz/.msh.h5
    meshing_session.meshing.File.ReadMesh(FileName=gz_path)
    meshing_session.meshing.File.ReadMesh(FileName=h5_path)

    # Step 4: Switch to solution and Write case file
    solver_session = meshing_session.switch_to_solver()
    solver_session.settings.solution.initialization.hybrid_initialize()
    gz_path = str(temporary_resource_dir / "default_case.cas.gz")
    h5_path = str(temporary_resource_dir / "default_case.cas.h5")
    write_case = solver_session.settings.file.write_case
    write_case(file_name=gz_path)
    write_case(file_name=h5_path)
    assert (temporary_resource_path / "default_case.cas.gz").exists()
    assert (temporary_resource_path / "default_case.cas.h5").exists()
    solver_session.exit()
    shutil.rmtree(temporary_resource_path, ignore_errors=True)


@pytest.mark.nightly
@pytest.mark.codegen_required
@pytest.mark.fluent_version("latest")
def test_case_file():
    case_path = examples.download_file(
        "elbow_param.cas.h5", "pyfluent/mixing_elbow", return_without_path=False
    )
    reader = CaseFile(case_file_name=case_path)

    assert reader.num_dimensions() == 3
    assert reader.precision() == 2

    plist = []
    olist = []
    for p in reader.input_parameters():
        ipar = OrderedDict()
        ipar["name"] = p.name
        ipar["numeric_value"] = p.numeric_value
        ipar["units"] = p.units
        plist.append(ipar)

    input_params = plist[0]
    assert input_params["name"] == "inlet2_temp"
    assert input_params["numeric_value"] == 500.0
    assert input_params["units"] == "K"

    for o in reader.output_parameters():
        opar = OrderedDict()
        opar["name"] = o.name
        opar["numeric_value"] = 0.0
        opar["units"] = o.units
        olist.append(opar)

    output_params = olist[0]
    assert output_params["name"] == "outlet_temp-op"
    assert output_params["numeric_value"] == 0.0
    assert output_params["units"] == "K"


@pytest.mark.nightly
@pytest.mark.codegen_required
@pytest.mark.fluent_version("latest")
def test_parameters(mixing_elbow_param_case_data_session):
    solver_session = mixing_elbow_param_case_data_session
    fluent_version = solver_session.get_fluent_version()
    input_params = solver_session.settings.parameters.input_parameters.list()
    for key, entry in input_params.items():
        input_value = entry[0]
        if fluent_version >= FluentVersion.v261:
            input_unit = list(entry[1].values())[0][-1]
        else:
            input_unit = entry[1]

    assert input_value == 500.0
    assert input_unit == "K"

    output_params = solver_session.settings.parameters.output_parameters.list()
    output_jdict = {}
    for key, entry in output_params.items():
        if fluent_version >= FluentVersion.v261:
            output_jdict[key] = list(entry[1].values())[0][-1]
        else:
            output_jdict[key] = entry[1]

    assert output_jdict == {"outlet_temp-op": "K"}


@pytest.mark.nightly
@pytest.mark.codegen_required
@pytest.mark.fluent_version("latest")
def test_parametric_project(mixing_elbow_param_case_data_session, new_solver_session):
    Path(pyfluent.config.examples_path).mkdir(parents=True, exist_ok=True)
    tmp_save_dir = Path(tempfile.mkdtemp(dir=pyfluent.config.examples_path)).parts[-1]
    init_project = Path(tmp_save_dir) / "mixing_elbow_param_init.flprj"
    project_file = Path(tmp_save_dir) / "mixing_elbow_param.flprj"

    session1 = mixing_elbow_param_case_data_session
    if not session1.connection_properties.inside_container:
        session1.chdir(pyfluent.config.examples_path)
    session1.settings.parametric_studies.initialize(project_filename=str(init_project))
    session1.settings.file.parametric_project.save_as(
        project_filename=str(project_file)
    )
    assert (Path(pyfluent.config.examples_path) / project_file).exists()

    session2 = new_solver_session
    if not session2.connection_properties.inside_container:
        session2.chdir(pyfluent.config.examples_path)
    session2.settings.file.parametric_project.open(project_filename=str(project_file))
    current_pstudy_name = session2.settings.current_parametric_study()
    assert current_pstudy_name == "elbow_param-Solve"
    pstudy = session2.settings.parametric_studies[current_pstudy_name]
    base_dp = pstudy.design_points["Base DP"]()
    base_inputs = base_dp["input_parameters"]
    assert base_inputs == {"inlet2_temp": 500.0}
    base_outputs = base_dp["output_parameters"]
    assert base_outputs == {"outlet_temp-op": pytest_approx(322.336008)}
    if session2.get_fluent_version() < FluentVersion.v251:
        pstudy.design_points.create_1()
        dp = pstudy.design_points["DP1"]
    else:
        dp = pstudy.design_points.create()
    dp.input_parameters["inlet2_temp"] = 600.0
    pstudy.design_points.update_selected(design_points=["DP1"])
    fluent_output_table = dp.output_parameters
    for key, entry in fluent_output_table.items():
        output_value_dp1 = entry()
    assert output_value_dp1 == pytest_approx(333.421)
